package com.at.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/29 9:24 PM
 */
object SparkStreaming06_State_Transform {

  def main(args: Array[String]): Unit = {
    // 创建环境
    // 创建时，需要传递2个参数：
    //    param： 环境配置
    val sc = new SparkConf().setMaster("local[*]").setAppName("operator")
    //    param2: 采集周期
    val ssc = new StreamingContext(sc, Seconds(3))

    val lines = ssc.socketTextStream("localhost", 9999)

    // transform 可以将底层RDD获取到后进行操作
    // 1、DStream功能不完善
    // 2、需要代码周期性执行
    //    val newDS = lines.transform(rdd => rdd.map(str => str))
    // Code: Driver
    val newDS = lines.transform(
      rdd => {
        // Code: Driver端,（周期性执行）
        rdd.map(
          str => {
            // Code: Executor
            str
          }
        )
      }
    )
//    val newDS1 = lines.map(data => data)
    // Code: Driver
    val newDS1 = lines.map(
      data => {
        // Executor
        data
      }
    )

    // 1、启动采集器
    ssc.start()
    // 2、等待采集器的关闭
    ssc.awaitTermination()
  }

}
